package com.eptok.gw.szcups.socket.client.agent;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.executor.ExecutorFilter;
import org.apache.mina.transport.socket.SocketSessionConfig;
import org.apache.mina.transport.socket.nio.NioSocketConnector;

import com.eptok.gw.szcups.common.GateWayConstant;

/**
 * @author:heqishan, Email:heqishan@gmail.com 2012-8-26 下午06:32:55
 */
public class ClientAgentAsynImpl implements ClientAgent, Reconnectable {
	
	private final NioSocketConnector connector;
	private final ExecutorService pool;
	private ClientConfig clientConfig;
	private IoSession session;
	
	public ClientAgentAsynImpl(ClientConfig clientConfig){
		connector = new NioSocketConnector();
		pool = Executors.newCachedThreadPool();
		this.clientConfig = clientConfig;
		init();
	}
	
	public void init() {
		connector.setConnectTimeoutMillis(GateWayConstant.ASYN_CONNECT_TIMEOUT);
		connector.getSessionConfig().setWriterIdleTime(GateWayConstant.ASYN_IDLE_TIME);
		connector.getFilterChain().addFirst("codec", new ProtocolCodecFilter(clientConfig.getCodecFactory()));
		connector.getFilterChain().addLast("threadPool", new ExecutorFilter(pool));
		connector.setHandler(clientConfig.getIoHandlerAdapter());
		
		SocketSessionConfig config = connector.getSessionConfig();
		config.setReceiveBufferSize(2048 * 50);
		config.setMaxReadBufferSize(2048 * 50);
		config.setMinReadBufferSize(2048 * 5);
		config.setSendBufferSize(2048 * 10);
		
		try{
			ConnectFuture future = connector.connect(new InetSocketAddress(clientConfig.getHost(), 
					clientConfig.getPort()));
			future.awaitUninterruptibly();
			session = future.getSession();
		}catch(Exception e){
			//Do nothing
//			e.printStackTrace();
		}
	}
	
	public boolean connect() {
		if(session != null && session.isConnected()){
			return true;
		}
		return false;
	}

	public void disconnect(){
		//Do nothing
	}
	
	private void shutdownAndAwaitTermination(final ExecutorService pool) {
		if (pool != null && !pool.isShutdown()) {
			while(! pool.isShutdown()){
				try {
					pool.shutdownNow();
					pool.awaitTermination(2, TimeUnit.SECONDS);
				} catch (final InterruptedException ie) {
					pool.shutdownNow();
				}
			}
		}
	}

	public Object recvMsg(String key) {
		return null;
	}

	public void sendMsg(Object object) {
		session.write(object).awaitUninterruptibly();
	}

	public void setRecvTimeout(int recvTimeout) {
		//do nothing
	}

	public boolean isConnected() {
		if(session != null && session.isConnected()){
			return true;
		}
		return false;
	}
	
	public synchronized void release() {
		if(session != null){
			session.close(false).awaitUninterruptibly();
//			session.getService().dispose();
		}
		shutdownAndAwaitTermination(pool);
		connector.dispose();
	}

	public boolean reconnect(){
		if(session != null && session.isConnected()){
			return true;
		}
		
		if(connector != null){
			if(connector.getHandler() == null){
				connector.setConnectTimeoutMillis(GateWayConstant.ASYN_CONNECT_TIMEOUT);
				connector.getSessionConfig().setBothIdleTime(GateWayConstant.ASYN_IDLE_TIME);
				connector.getFilterChain().addFirst("codec", new ProtocolCodecFilter(clientConfig.getCodecFactory()));
				connector.getFilterChain().addLast("threadPool", new ExecutorFilter(pool));
				connector.setHandler(clientConfig.getIoHandlerAdapter());
			}
			
			try{
				ConnectFuture future = connector.connect(new InetSocketAddress(clientConfig.getHost(), 
						clientConfig.getPort()));
				future.awaitUninterruptibly();
				session = future.getSession();
				return true;
			}catch(Exception e){
//				e.printStackTrace();
			}
			
			if(session != null && session.isConnected()){
				session.close(false).awaitUninterruptibly();
				session.getService().dispose();
			}
		}
		return false;
	}
	
}
